package com.erik.lin.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * Created by Erik on 2017/7/27.
 */
public class DataExKafkaProcuder extends Thread {

    private String topic;

    public DataExKafkaProcuder(String topic){
        super();
        this.topic = topic;
    }


    @Override
    public void run() {
        KafkaProducer producer = createProducer();
        while(true){
            try {
                String str = "event1^13080624"+String.valueOf(new Random().nextInt(1000));
                producer.send(new ProducerRecord<String, byte[]>(topic,
                        strToByteArray(str))).get();
                System.out.println("MSG:" + str);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    private KafkaProducer createProducer() {
        Map<String,Object> config=new HashMap<String, Object>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.8.8:9092,192.168.8.8:9093,192.168.8.8:9094");// 声明kafka broker ，要注意地址一定要正确
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
        config.put(ProducerConfig.LINGER_MS_CONFIG, 0);
        return new KafkaProducer<String, String>(config);
    }

    public static byte[] strToByteArray(String str) {
        if (str == null) {
            return null;
        }
        byte[] byteArray = str.getBytes();
        return byteArray;
    }

    public static void main(String[] args) {
        new DataExKafkaProcuder("test").start();// 使用kafka集群中创建好的主题 test
    }

}
